package lb.io.lblet;

import io.grpc.stub.StreamObserver;
import lb.io.api.core.v1.*;
import lb.io.api.event.v1.CarryOut;
import lb.io.api.event.v1.CloneOut;
import lb.io.api.event.v1.OracleOut;
import lb.io.api.system.v1.Lblet;
import lb.io.apimachinery.pkg.apis.meta.v1.Status;
import lb.io.apis.meta.v1.ResourceStatus;
import lb.io.apis.meta.v1.codes.Codez;
import lb.io.apis.meta.v1.codes.LBStatus;
import lb.io.apis.meta.v1.codes.LetException;
import lb.io.lblet.pkg.service.pb.*;
import lb.io.object.Tool;
import lb.io.options.LetOptions;
import lb.io.utils.Klog;
import lb.io.utils.Progress;
import lb.io.utils.SystemUtil;
import org.apache.logging.log4j.Logger;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.nio.file.Paths;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

// 实现服务接口的类
class Let extends LbletGrpc.LbletImplBase {

    // 日志
    private static final Logger klog = Klog.getLogger(LbLetServer.class);

    private static final Status ok = Status.newBuilder().setCode(0).setMessage("Success").build();

    private LetOptions letOptions = null;

    private Lblet lblet = null;

    public Let(Lblet lblet, LetOptions options) {
        this.lblet = lblet;
        this.letOptions = options;
    }

    @Override
    public void healthz(HealthzRequest request, StreamObserver<HealthzResponse> responseObserver) {
        HealthzResponse response = HealthzResponse.newBuilder().setMessage("pong").build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public StreamObserver<UpgradeRequest> upgrade(StreamObserver<UpgradeResponse> responseObserver) {
        return new StreamObserver<UpgradeRequest>() {
            @Override
            public void onNext(UpgradeRequest upgradeRequest) {

            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {

            }
        };
    }

    @Override
    public StreamObserver<SSHRequest> webSSH(StreamObserver<SSHResponse> responseObserver) {
        return new StreamObserver<SSHRequest>() {
            @Override
            public void onNext(SSHRequest sshRequest) {

            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {

            }
        };
    }

    @Override
    public void pingz(PingzRequest request, StreamObserver<PingzResponse> responseObserver) {
        PingzResponse response = PingzResponse.newBuilder().setMessage("pong").setStatus(ok).build();
        // onNext() 方法向客户端返回结果
        responseObserver.onNext(response);
        // 告诉客户端这次调用已经完成
        responseObserver.onCompleted();
    }

    @Override
    public void getVcHosts(GetVcHostsRequest request, StreamObserver<GetVcHostsResponse> responseObserver) {
        GetVcHostsResponse response = GetVcHostsResponse.newBuilder().setStatus(ok).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getCapacity(GetCapacityRequest request, StreamObserver<GetCapacityResponse> responseObserver) {
        GetCapacityResponse response = GetCapacityResponse.newBuilder().build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getVMsByDatacenter(GetVMsByDatacenterRequest request, StreamObserver<GetVMsByDatacenterResponse> responseObserver) {
        GetVMsByDatacenterResponse response = GetVMsByDatacenterResponse.newBuilder().setStatus(ok).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getVmwareInfo(GetVmwareInfoRequest request, StreamObserver<GetVmwareInfoResponse> responseObserver) {
        GetVmwareInfoResponse response = GetVmwareInfoResponse.newBuilder().setStatus(ok).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getVmwareMachines(GetVmwareMachinesRequest request, StreamObserver<GetVmwareMachinesResponse> responseObserver) {
        GetVmwareMachinesResponse response = GetVmwareMachinesResponse.newBuilder().setStatus(ok).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getDatacenter(GetDatacenterRequest request, StreamObserver<GetDatacenterResponse> responseObserver) {
        GetDatacenterResponse response = GetDatacenterResponse.newBuilder().setStatus(ok).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getClusterHost(GetClusterHostRequest request, StreamObserver<GetClusterHostResponse> responseObserver) {
        GetClusterHostResponse response = GetClusterHostResponse.newBuilder().setStatus(ok).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getVcenterFolder(GetVcenterFolderRequest request, StreamObserver<GetVcenterFolderResponse> responseObserver) {
        GetVcenterFolderResponse response = GetVcenterFolderResponse.newBuilder().setStatus(ok).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getVcenterHosts(GetVcenterHostsRequest request, StreamObserver<GetVcenterHostsResponse> responseObserver) {
        GetVcenterHostsResponse response = GetVcenterHostsResponse.newBuilder().setStatus(ok).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getVcenterResourcePool(GetVcenterResourcePoolRequest request, StreamObserver<GetVcenterResourcePoolResponse> responseObserver) {
        GetVcenterResourcePoolResponse response = GetVcenterResourcePoolResponse.newBuilder().setStatus(ok).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getMssqlDatabase(GetMssqlDatabaseRequest request, StreamObserver<GetMssqlDatabaseResponse> responseObserver) {
        GetMssqlDatabaseResponse response = GetMssqlDatabaseResponse.
                newBuilder().
                setStatus(ok).
                build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getMysqlDatabase(GetMysqlDatabaseRequest request, StreamObserver<GetMysqlDatabaseResponse> responseObserver) {
        GetMysqlDatabaseResponse response = GetMysqlDatabaseResponse.
                newBuilder().
                setStatus(ok).
                build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getFileSystemNextPath(FileSystemNextPathRequest request, StreamObserver<FileSystemNextPathResponse> responseObserver) {
        FileSystemNextPathResponse response = FileSystemNextPathResponse.
                newBuilder().
                setStatus(ok).
                build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    private String testApplication(Application app) throws LetException {
        String result = "";
        boolean connected = false;
        for (OracleInstance instance : app.getSpec().getOracle().getInstancesList()) {
            String dbStatus = LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                    .Op("getDBStatus")
                    .Sid(instance.getSid())
                    .User(instance.getUser())
                    .Home(instance.getHome())
                    .Listen(instance.getListen()).Execute();
            if (!dbStatus.trim().equals("TRUE")) {
                connected = true;
                break;
            }
        }
        if (!connected) {
            throw new LetException(Codez.ErrOracleConnect, "application instance: connection failed");
        }
        OracleInstance instance = app.getSpec().getOracle().getInstances(0);
        String archiveLog = LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                .Op("getDBLogMode")
                .Sid(instance.getSid())
                .User(instance.getUser())
                .Home(instance.getHome())
                .Listen(instance.getListen()).Execute();
        if (!archiveLog.toLowerCase().trim().equals("archivelog")) {
            result = "error";
            throw new LetException(Codez.ErrOracleNoArchive, "未打开归档日志 archivelog");
        }

        return result;
    }

    @Override
    public void testApplication(TestApplicationRequest request, StreamObserver<TestApplicationResponse> responseObserver) {
        String result = "";
        Status status = LBStatus.OK();

        try {
            result = this.testApplication(request.getApp());
        } catch (LetException ex) {
            status = new LBStatus(ex).getStatus();
        }

        TestApplicationResponse response = TestApplicationResponse.newBuilder()
                .setResult(result)
                .setStatus(status)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void testPod(TestPodRequest request, StreamObserver<TestPodResponse> responseObserver) {
        Status status = LBStatus.OK();
        String result = "ok";
        TestPodResponse response = TestPodResponse.newBuilder()
                .setResult(result)
                .setStatus(status)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getOracleEnv(GetOracleEnvRequest request, StreamObserver<GetOracleEnvResponse> responseObserver) {
        String asm = LBKit.Get.OracleASM();
        OracleAttributes asmAttr = null;
        if (!asm.equals("")) {
            String home = LBKit.Get.OracleHome(asm);
            String listen = "";
            if (!home.equals("")) {
                listen = Paths.get(home, "network", "admin").toString();
            }
            asmAttr = OracleAttributes.newBuilder()
                    .setDb(asm)
                    .setSid(asm)
                    .setUser(LBKit.Get.ASMUser(asm))
                    .setHome(home)
                    .setListen(listen)
                    .build();
        }
        OracleAttributes oracleAttributes = null;
        {
            String user = "oracle";
            String home = LBKit.Get.OracleHome("");
            String listen = LBKit.Get.TNSAdmin(user);
            oracleAttributes = OracleAttributes.newBuilder()
                    .setUser(user)
                    .setHome(home)
                    .setListen(listen)
                    .build();
        }
        GetOracleEnvResponse response = GetOracleEnvResponse.newBuilder()
                .setOracle(oracleAttributes)
                .setStatus(LBStatus.OK())
                .build();
        if (asmAttr != null) {
            response = response.toBuilder().setAsm(asmAttr).build();
        }
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getOracleInstances(GetOracleInstancesRequest request, StreamObserver<GetOracleInstancesResponse> responseObserver) {
        List<OracleInstance> instances = new LinkedList<>();
        String asm = LBKit.Get.OracleASM();
        OracleAttributes asmAttr = null;
        if (!asm.equals("")) {
            String home = LBKit.Get.OracleHome(asm);
            String listen = "";
            if (!home.equals("")) {
                listen = Paths.get(home, "network", "admin").toString();
            }
            asmAttr = OracleAttributes.newBuilder()
                    .setDb(asm)
                    .setSid(asm)
                    .setUser(LBKit.Get.ASMUser(asm))
                    .setHome(home)
                    .setListen(listen)
                    .build();
        }
        OracleInstance instance = null;
        List<String> sids = LBKit.Get.OracleSid();
        for (String sid : sids) {
            if (sid.equals("")) {
                continue;
            }
            String _sid = sid;
            Pattern compile = Pattern.compile("([a-zA-Z]+)(\\d+)");
            Matcher matcher = compile.matcher(_sid);
            if (matcher.find()) {
                _sid = matcher.group(1);
            }
            String home = LBKit.Get.OracleHome(_sid);
            String user = LBKit.Get.OracleUser(sid);
            String listen = LBKit.Get.TNSAdmin(user);
            if (listen.equals("")) {
                listen = Paths.get(home, "network", "admin").toString();
            }
            String db = LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                    .Op("getDBName")
                    .Sid(sid)
                    .User(user)
                    .Home(home)
                    .Listen(listen)
                    .Execute();
            long size = Long.parseLong(LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                    .Op("getDBSize")
                    .Sid(sid)
                    .User(user)
                    .Home(home)
                    .Listen(listen)
                    .Execute());
            instance = OracleInstance.newBuilder()
                    .setDb(db.toLowerCase())
                    .setSid(sid)
                    .setUser(user)
                    .setHome(home)
                    .setListen(listen)
                    .setDbSize(size)
                    .build();
            if (asmAttr != null) {
                instance = instance.toBuilder().setAsm(asmAttr).build();
            }
            instances.add(instance);
        }
        GetOracleInstancesResponse response = GetOracleInstancesResponse.newBuilder()
                .addAllInstances(instances)
                .setStatus(LBStatus.OK())
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void testOracleInstance(TestOracleInstanceRequest request, StreamObserver<TestOracleInstanceResponse> responseObserver) {
        OracleInstance instance = request.getInstance();
        String openMode = LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                .Op("getOpenMode")
                .Sid(instance.getSid())
                .User(instance.getUser())
                .Home(instance.getHome())
                .Listen(instance.getListen()).Execute();
        String msg = openMode.trim().equals("OPEN") ? "success" : "failed";
        TestOracleInstanceResponse response = TestOracleInstanceResponse.newBuilder()
                .setMsg(msg)
                .setStatus(LBStatus.OK())
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void getOracleParameters(GetOracleParametersRequest request, StreamObserver<GetOracleParametersResponse> responseObserver) {
        OracleInstance instance = request.getInstance();
        LBKit.Sqlplus sqlplus = LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                .Sid(instance.getSid())
                .User(instance.getUser())
                .Home(instance.getHome())
                .Listen(instance.getListen());
        GetOracleParametersResponse response = GetOracleParametersResponse.newBuilder()
                .setParam(OracleParam.newBuilder()
                        .setDbSize(sqlplus.Op("getDBSize").Execute())
                        .setVersion(sqlplus.Op("getDBVersion").Execute())
                        .setBct(sqlplus.Op("getBCTStatus").Execute())
                        .setArchive(sqlplus.Op("getArchiveDest").Execute())
                        .build())
                .setStatus(LBStatus.OK())
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    private PersistentVolumeClaim createClaim(PersistentVolumeClaim claim) throws LetException {
        ClaimProtocol pc = claim.getSpec().getClaim();
        if (!pc.getProtocol().equals("iscsi")) {
            throw new LetException(Codez.ErrISCSI, "only support iscsi protocol");
        }
        String disk = new LBKit.iscsi(letOptions.getScripts(), pc.getHost(), pc.getBindID()).start();
        if (disk.equals("")) {
            throw new LetException(Codez.ErrISCSI, "连接 iscsi 失败");
        }
        claim = claim.toBuilder()
                .setStatus(claim.getStatus().toBuilder().setDevice(disk).build())
                .build();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return claim;
    }

    @Override
    public void createClaim(CreateClaimRequest request, StreamObserver<CreateClaimResponse> responseObserver) {
        Status status = LBStatus.OK();
        PersistentVolumeClaim claim = request.getPvc();

        try {
            claim = this.createClaim(claim);
        } catch (LetException ex) {
            status = new LBStatus(ex).getStatus();
        }

        CreateClaimResponse response = CreateClaimResponse.newBuilder()
                .setClaim(claim)
                .setStatus(status)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    private PersistentVolumeClaim deleteClaim(PersistentVolumeClaim claim) throws LetException {
        ClaimProtocol cp = claim.getSpec().getClaim();
        if (!cp.getProtocol().equals("iscsi")) {
            throw new LetException(Codez.ErrISCSI, "only support iscsi protocol");
        }
        new LBKit.iscsi(letOptions.getScripts(), cp.getHost(), cp.getBindID()).stop();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            klog.error(e.getMessage());
        }
        return claim;
    }

    @Override
    public void deleteClaim(DeleteClaimRequest request, StreamObserver<DeleteClaimResponse> responseObserver) {
        PersistentVolumeClaim claim = request.getClaim();
        Status status = LBStatus.OK();
        try {
            claim = this.deleteClaim(claim);
        } catch (LetException ex) {
            status = new LBStatus(ex).getStatus();
        }

        DeleteClaimResponse response = DeleteClaimResponse.newBuilder()
                .setClaim(claim)
                .setStatus(status)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    private Pod createPod(Pod pod, List<PersistentVolumeClaim> claims) throws LetException {
        if (claims.size() > 0 && pod.getSpec().getModule().getModule().equals("oracle")) {
            PodDiskStatus diskStatus = null;
            if (pod.getSpec().getModule().getOracle().hasAsm()) {
                diskStatus = disk.createDiskGroup(letOptions.getScripts(), pod, claims);
            } else {
                String dev = claims.get(0).getStatus().getDevice();
                String point = Paths.get(letOptions.getPointDir(), pod.getMetadata().getName()).toString();
                diskStatus = disk.createFileDisk(letOptions.getScripts(), dev, point);

                try {
                    String file = Paths.get(point, "linkBackup").toString();
                    SystemUtil.writeAll(file, "");
                } catch (IOException e) {
                    throw new LetException(Codez.ErrCreateFileDisk, e.getMessage());
                }
            }

            pod = pod.toBuilder()
                    .setMetadata(pod.getMetadata().toBuilder()
                            .setOwnerComponent(Tool.buildOwnerComponent(lblet))
                            .build())
                    .setStatus(pod.getStatus().toBuilder()
                            .setPodDisk(diskStatus)
                            .setCreateTime(System.currentTimeMillis() / 1000)
                            .setOs(OSStatus.newBuilder()
                                    .setKernel(lblet.getStatus().getOs().getKernel())
                                    .setProcesser(lblet.getStatus().getOs().getProcesser())
                                    .build())
                            .setStatus(ResourceStatus.OKState.getStatus())
                            .setRuntime("idle")
                            .setIp(lblet.getSpec().getIp().getIp())
                            .build())
                    .build();

        }

        return pod;
    }

    @Override
    public void createPod(CreatePodRequest request, StreamObserver<CreatePodResponse> responseObserver) {
        Pod pod = request.getPod();
        Status status = LBStatus.OK();
        List<PersistentVolumeClaim> claims = request.getClaimsList();

        try {
            pod = this.createPod(pod, claims);
        } catch (LetException ex) {
            status = new LBStatus(ex).getStatus();
        }

        CreatePodResponse response = CreatePodResponse.newBuilder()
                .setPod(pod)
                .setStatus(status)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
        // 创建Pod流程结束，关闭线程
        klog.info("create <Pod:{}:{}> completed", pod.getMetadata().getNamespace(), pod.getMetadata().getName());
    }

    private Pod rebuildPod(Pod pod, List<PersistentVolumeClaim> claims) throws LetException {
        if (claims.size() > 0 && pod.getSpec().getModule().getModule().equals("oracle")) {
            PodDiskStatus diskStatus = null;
            if (pod.getSpec().getModule().getOracle().hasAsm()) {
                diskStatus = disk.renameDiskGroup(letOptions.getScripts(), pod, claims);
                String store = Paths.get(letOptions.getPointDir(), pod.getMetadata().getName()).toString();
                diskStatus = diskStatus.toBuilder().setStore(store).build();
            } else {
                String dev = claims.get(0).getStatus().getDevice();
                String point = Paths.get(letOptions.getPointDir(), pod.getMetadata().getName()).toString();
                diskStatus = disk.renameFileDisk(letOptions.getScripts(), dev, point);
            }
            String protocl = claims.get(0).getSpec().getClaim().getProtocol();
            String content = "";
            switch (protocl) {
                case "iscsi":
                    content = lblet.getStatus().getWwn();
                    break;
                case "fc":
                    content = lblet.getStatus().getInitiatorName();
                    break;
            }
            diskStatus = diskStatus.toBuilder()
                    .setSource(diskStatus.getSource().toBuilder()
                            .setDiskProtocol(protocl)
                            .setContent(content)
                            .build())
                    .build();

            pod = pod.toBuilder()
                    .setMetadata(pod.getMetadata().toBuilder()
                            .putAnnotations("pod", "clone")
                            .setOwnerComponent(Tool.buildOwnerComponent(lblet))
                            .build())
                    .setStatus(pod.getStatus().toBuilder()
                            .setPodDisk(diskStatus)
                            .setCreateTime(System.currentTimeMillis() / 1000)
                            .setOs(OSStatus.newBuilder()
                                    .setKernel(lblet.getStatus().getOs().getKernel())
                                    .setProcesser(lblet.getStatus().getOs().getProcesser())
                                    .build())
                            .setStatus(ResourceStatus.OKState.getStatus())
                            .setRuntime("rebuild")
                            .setIp(lblet.getSpec().getIp().getIp())
                            .build())
                    .build();

        }

        return pod;
    }

    @Override
    public void rebuildPod(RebuildPodRequest request, StreamObserver<RebuildPodResponse> responseObserver) {
        Pod pod = request.getPod();
        List<PersistentVolumeClaim> claims = request.getClaimsList();
        Status status = LBStatus.OK();

        try {
            pod = this.rebuildPod(pod, claims);
        } catch (LetException ex) {
            status = new LBStatus(ex).getStatus();
        }

        RebuildPodResponse response = RebuildPodResponse.newBuilder()
                .setPod(pod)
                .setStatus(status)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
        // 创建Pod流程结束，关闭线程
        klog.info("rebuild <Pod:{}:{}> completed", pod.getMetadata().getNamespace(), pod.getMetadata().getName());
    }

    private Pod deletePod(Pod pod) throws LetException {
        if (pod.getMetadata().getAnnotationsMap().containsKey("pod")
                && pod.getMetadata().getAnnotationsMap().get("pod").equals("clone")) {
            OraclePodSpec oracle = pod.getSpec().getModule().getOracle();
            final String db = oracle.getInstance().getDb();
            String sid = oracle.getInstance().getSid();
            String user = oracle.getInstance().getUser();
            String home = oracle.getInstance().getHome();
            String listen = oracle.getInstance().getListen();
            String rlog = Paths.get(letOptions.getPointDir(),
                    String.format("%s-recover.log", sid)).toString();
            String params = Paths.get(letOptions.getPointDir(),
                    String.format("%s-param", sid)).toString();
            String target = pod.getStatus().getPodDisk().getTarget();
            String store = pod.getStatus().getPodDisk().getStore();
            LBKit.Recover recover = LBKit.Recover.RecoverBuilder(letOptions.getScripts())
                    .Op("delete")
                    .Sid(sid)
                    .Target(target)
                    .Store(store)
                    .Rlog(rlog)
                    .User(user)
                    .Home(home)
                    .Listen(listen);
            if (oracle.hasAsm()) {
                recover.EnableASM();
            }
            recover.Execute();
            new File(rlog).delete();
            new File(params).delete();
            rlog = Paths.get(letOptions.getPointDir(),
                    String.format("%s-backup.log", db)).toString();
            new File(rlog).delete();
        }
        if (pod.getStatus().hasPodDisk()) {
            PodDiskStatus diskStatus = pod.getStatus().getPodDisk();
            switch (diskStatus.getKind()) {
                case "AIXFS":
                    String point = diskStatus.getAixFS().getPath();
                    String dev = diskStatus.getAixFS().getPv();
                    diskStatus = disk.deleteFileDisk(letOptions.getScripts(), dev, point);
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {

                    }
                    break;
                case "AIXRaw":
                    AIXRaw raw = diskStatus.getAixRaw();
                    OraclePodSpec oracle = pod.getSpec().getModule().getOracle();
                    if (oracle.hasAsm()) {
                        OracleAttributes asm = oracle.getAsm();
                        OracleAttributes instance = oracle.getInstance();

                        try {
                            String diskgroup = new LBKit.diskgroup(letOptions.getScripts(), raw.getGroup(), asm.getSid(), null)
                                    .User(instance.getUser())
                                    .Home(instance.getHome())
                                    .Listen(instance.getListen())
                                    .delete(true);
                        } catch (RuntimeException ex) {
                            klog.error(ex.getMessage());
                        }

                        try {
                            Thread.sleep(5000);
                        } catch (InterruptedException e) {

                        }

                        for (AIXRawItem item : raw.getRawsList()) {
                            new LBKit.rawDisk(letOptions.getScripts(), item.getDev(), "", "").delete();
                            try {
                                Thread.sleep(5000);
                            } catch (InterruptedException e) {

                            }
                        }
                    }
                    break;
            }
        }

        return pod;
    }

    @Override
    public void deletePod(DeletePodRequest request, StreamObserver<DeletePodResponse> responseObserver) {
        Pod pod = request.getPod();
        Status status = LBStatus.OK();

        try {
            pod = this.deletePod(pod);
        } catch (LetException ex) {
            status = new LBStatus(ex).getStatus();
        }

        DeletePodResponse response = DeletePodResponse.newBuilder()
                .setPod(pod)
                .setStatus(status)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public StreamObserver<BackupRequest> backup(final StreamObserver<BackupResponse> responseObserver) {
        return new StreamObserver<BackupRequest>() {

            private final Channel<Message> channel = new Channel<>(1);

            private boolean cancel = false;

            public synchronized boolean isCancel() {
                return cancel;
            }

            public synchronized void setCancel(boolean cancel) {
                this.cancel = cancel;
            }

            private void Do(Pod pod, final CarrySpec body, final StreamObserver<BackupResponse> responseObserver) throws LetException {
                if (!pod.getSpec().getModule().getModule().equals("oracle")) {
                    throw new LetException(Codez.ErrBadParameters, "aix 只支持 oracle 备份");
                }


                final String clientID = body.getClientID();
                klog.info("backup task:{} start...", clientID);
                final OraclePodSpec oracle = pod.getSpec().getModule().getOracle();
                String target;
                String store;

                // 备份开始前先挂载磁盘
                PodDiskStatus disk = pod.getStatus().getPodDisk();
                switch (disk.getKind()) {
                    case "AIXRaw":
                        String diskGroup = disk.getAixRaw().getGroup();
                        target = disk.getTarget();
                        store = disk.getStore();
                        SystemUtil.mkdir(store);
                        try {
                            new LBKit.diskgroup(letOptions.getScripts(), diskGroup, oracle.getAsm().getSid(), null)
                                    .User(oracle.getAsm().getUser())
                                    .Home(oracle.getAsm().getHome())
                                    .Listen(oracle.getAsm().getListen())
                                    .connect();
                        } catch (RuntimeException e) {
                            throw new LetException(Codez.ErrConnectDiskGroup, e.getMessage());
                        }

                        break;
                    case "AIXFS":
                        String dev = disk.getAixFS().getPv();
                        target = disk.getTarget();
                        store = target;
                        try {
                            new LBKit.fsDisk(letOptions.getScripts(), target, dev).connect();
                        } catch (RuntimeException e) {
                            throw new LetException(Codez.ErrConnectFileDisk, e.getMessage());
                        }

                        if (!new File(Paths.get(target, "linkBackup").toString()).exists()) {
                            throw new LetException(Codez.ErrMountUnknown, "未发现标志文件");
                        }

                        break;
                    default:
                        target = disk.getTarget();
                        store = target;
                }

                String db = oracle.getInstance().getDb();
                String sid = oracle.getInstance().getSid();
                String user = oracle.getInstance().getUser();
                String home = oracle.getInstance().getHome();
                String listen = oracle.getInstance().getListen();
                final String rlog = Paths.get(letOptions.getPointDir(), String.format("%s-backup.log", db)).toString();
                try {
                    SystemUtil.truncate(rlog);
                } catch (IOException e) {
                    klog.warn(e.getMessage());
                    throw new LetException(Codez.ErrLetNotExists, e.getMessage());
                }

                Date now = new Date();
                long from = body.getModule().getOracle().getLastTime();
                if (from <= 0) {
                    from = now.getTime() / 1000 - 1;
                }
                while (from > now.getTime() / 1000) {
                    from = from - 3600 * 8;
                }
                channel.put(new Message("+start", "", 0));

                final String storePath = store;
                final OracleAttributes tempOracle = oracle.getInstance();
                // 另起一个线程，用来发送备份数据
                final Thread senderT = new Thread() {
                    @Override
                    public void run() {
                        CarryOut out = CarryOut.newBuilder().setClientID(clientID).build();
                        Message msg = null;
                        boolean over = false;
                        while (true) {
                            if (isInterrupted()) {
                                break;
                            }

                            if (over) {
                                break;
                            }

                            if (isCancel()) {
                                break;
                            }

                            msg = channel.take();
                            if (msg == null) {
                                continue;
                            }

                            switch (msg.getPhase()) {
                                case "+start":
                                    out = out.toBuilder()
                                            .setStartTime(msg.getTimestamp())
                                            .setPhase(msg.getPhase())
                                            .build();
                                    send(out);
                                    break;
                                case "+continue":
                                    String iops = LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                                            .Op("getRmanJobDetails")
                                            .Sid(tempOracle.getSid())
                                            .User(tempOracle.getUser())
                                            .Home(tempOracle.getHome())
                                            .Listen(tempOracle.getListen()).Execute();

                                    long read = 0;
                                    long write = 0;
                                    double bw = 0;
                                    double process = 0;
                                    if (!iops.equals("")) {
                                        String[] sp = iops.split(" ");
                                        List<String> line = new ArrayList<>();
                                        for (String item : sp) {
                                            if (!item.trim().equals("")) line.add(item);
                                        }
                                        if (line.size() > 2) {
                                            read = Long.parseLong(line.get(0));
                                            write = Long.parseLong(line.get(1));
                                            bw = Double.parseDouble(line.get(2));
                                        }
                                        String processString = LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                                                .Op("getBackupProcess")
                                                .Sid(tempOracle.getSid())
                                                .User(tempOracle.getUser())
                                                .Home(tempOracle.getHome())
                                                .Listen(tempOracle.getListen()).Execute();
                                        if (!processString.equals("")) {
                                            process = Double.parseDouble(processString);
                                            BigDecimal b = new BigDecimal(process);
                                            process = b.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
                                        }
                                    }
                                    out = out.toBuilder()
                                            .setPhase(msg.getPhase())
                                            .setRead(read)
                                            .setWrite(write)
                                            .setBandWidth(bw)
                                            .setProcess(process)
                                            .build();
                                    send(out);
                                    break;
                                case "+failed":
                                    String rlogData = null;
                                    try {
                                        String readData = SystemUtil.readAll(rlog);
                                        rlogData = SystemUtil.base64Encode(readData);
                                    } catch (Exception e) {
                                        klog.error("read rlog failed: {}", e.getMessage());
                                    }
                                    out = out.toBuilder()
                                            .setProcess(100)
                                            .setPhase(msg.getPhase())
                                            .setEndTime(msg.getTimestamp())
                                            .setLog(rlogData)
                                            .setErr(msg.getError())
                                            .build();
                                    send(out);
                                    over = true;
                                    break;
                                case "+completed":
                                    String rlogData1 = null;
                                    try {
                                        String readData = SystemUtil.readAll(rlog);
                                        rlogData1 = SystemUtil.base64Encode(readData);
                                    } catch (Exception e) {
                                        klog.error("read rlog failed: {}", e.getMessage());
                                    }
                                    String controlfile = "";
                                    String pfile = "";
                                    OracleOut oracleOut = null;
                                    if (oracle.hasAsm() && body.getModule().getOracle().getOp().equals("backupDB")) {
                                        String controlfilePath = Paths.get(storePath, "controlfile.trace").toString();
                                        String pfilePath = Paths.get(storePath, tempOracle.getDb() + "__backup.ora").toString();
                                        try {
                                            controlfile = SystemUtil.base64Encode(SystemUtil.readAll(controlfilePath));
                                            pfile = SystemUtil.base64Encode(SystemUtil.readAll(pfilePath));
                                        } catch (UnsupportedEncodingException e) {
                                            klog.error(e.getMessage());
                                        }
                                        oracleOut = OracleOut.newBuilder()
                                                .setControlfile(controlfile)
                                                .setPfile(pfile)
                                                .build();
                                    }
                                    out = out.toBuilder()
                                            .setProcess(100)
                                            .setPhase(msg.getPhase())
                                            .setEndTime(msg.getTimestamp())
                                            .setLog(rlogData1)
                                            .setErr(msg.getError())
                                            .build();
                                    if (oracleOut != null) {
                                        out = out.toBuilder().setOracle(oracleOut).build();
                                    }
                                    send(out);
                                    over = true;
                                    break;
                            }
                        }
                    }

                    private void send(CarryOut out) {
                        BackupResponse response = BackupResponse.newBuilder()
                                .setOut(out)
                                .setStatus(LBStatus.OK())
                                .build();
                        if (!isCancel()) {
                            responseObserver.onNext(response);
                        }
                    }
                };
                senderT.start();

                // 另起一个线程，用来确定备份情况
                final Thread healthT = new Thread() {
                    @Override
                    public void run() {
                        while (!isInterrupted() && !isCancel()) {
                            channel.put(new Message("+continue", "", 0));
                            try {
                                sleep(10000);
                            } catch (InterruptedException e) {
                            }
                        }
                    }
                };

                LBKit.Rman rman = LBKit.Rman.RmanBuilder(letOptions.getScripts())
                        .DB(db)
                        .Sid(sid)
                        .Target(target)
                        .Store(store)
                        .Rlog(rlog)
                        .User(user)
                        .Home(home)
                        .Listen(listen);
                if (oracle.hasAsm()) {
                    rman.EnableASM();
                }

                String format = "yyyy-MM-dd..HH:mm:ss";
                healthT.start();
                switch (body.getModule().getOracle().getOp()) {
                    case "backupDB":
                        // 先执行 rman check
                        klog.info("Pod <{}:{}> check rman", pod.getMetadata().getNamespace(), pod.getMetadata().getName());
                        String result = rman.Op("check").Execute();
                        if (result.equals("")) {
                            healthT.interrupt();
                            channel.put(new Message("+failed", "rman检测错误", now.getTime()));
                            break;
                        }

                        // 开始备份数据
                        klog.info("Pod <{}:{}> backup rman", pod.getMetadata().getNamespace(), pod.getMetadata().getName());
                        result = rman.Op("backup").Execute();
                        if (result.equals("")) {
                            healthT.interrupt();
                            channel.put(new Message("+failed", "rman数据备份错误", now.getTime()));
                            break;
                        }

                        // 不加 break，数据备份结束后备份日志
                    case "backupLOG":

                        String role = LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                                .Op("getDBRole")
                                .Sid(oracle.getInstance().getSid())
                                .User(oracle.getInstance().getUser())
                                .Home(oracle.getInstance().getHome())
                                .Listen(oracle.getInstance().getListen()).Execute();
                        if (!role.trim().equals("") && !role.equals("standby")) {
                            LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                                    .Op("switchArchiveLog")
                                    .Sid(oracle.getInstance().getSid())
                                    .User(oracle.getInstance().getUser())
                                    .Home(oracle.getInstance().getHome())
                                    .Listen(oracle.getInstance().getListen()).Execute();
                        }

                        String fromTime = new SimpleDateFormat(format).format(new Date(from * 1000));
                        now = new Date();
                        String utilTime = new SimpleDateFormat(format).format(now);
                        klog.info("Pod <{}:{}> log rman", pod.getMetadata().getNamespace(), pod.getMetadata().getName());
                        result = rman.Op("log").fromTime(fromTime).utilTime(utilTime).Execute();
                        if (result.equals("")) {
                            healthT.interrupt();
                            channel.put(new Message("+failed", "rman日志备份错误", now.getTime() / 1000));
                            break;
                        }

                        healthT.interrupt();
                        channel.put(new Message("+completed", "", now.getTime() / 1000));
                        break;
                    default:
                        healthT.interrupt();
                        channel.put(new Message("+failed", "invalid backup operation", now.getTime() / 1000));
                        break;
                }

                klog.info("backup task:{} completed", clientID);

                // 配置结束，卸载磁盘
                switch (disk.getKind()) {
                    case "AIXRaw":
                        String diskGroup = disk.getAixRaw().getGroup();
                        new LBKit.diskgroup(letOptions.getScripts(), diskGroup, sid, null)
                                .User(oracle.getAsm().getUser())
                                .Home(oracle.getAsm().getHome())
                                .Listen(oracle.getAsm().getListen())
                                .disconnect();
                        break;
                    case "AIXFS":
                        String dev = disk.getAixFS().getPv();
                        new LBKit.fsDisk(letOptions.getScripts(), target, dev).disconnect();
                        break;
                }

                // 备份流程结束，不再发送数据
                senderT.interrupt();
            }

            @Override
            public void onNext(final BackupRequest request) {
                Pod pod = request.getPod();

                try {
                    this.Do(request.getPod(), request.getBody(), responseObserver);
                } catch (LetException ex) {
                    Status status = new LBStatus(ex).getStatus();
                    BackupResponse response = BackupResponse.newBuilder()
                            .setStatus(status)
                            .build();
                    responseObserver.onNext(response);
                    this.onCompleted();
                }
            }

            @Override
            public void onError(Throwable throwable) {
                responseObserver.onError(throwable);
            }

            @Override
            public void onCompleted() {
                setCancel(true);
                responseObserver.onCompleted();
            }
        };
    }

    @Override
    public StreamObserver<RecoverRequest> recover(final StreamObserver<RecoverResponse> responseObserver) {
        return new StreamObserver<RecoverRequest>() {

            private boolean cancel = false;

            public synchronized boolean isCancel() {
                return cancel;
            }

            public synchronized void setCancel(boolean cancel) {
                this.cancel = cancel;
            }

            private void Do(Pod pod, final CloneSpec body, final StreamObserver<RecoverResponse> responseObserver) throws LetException {
                if (!pod.getSpec().getModule().getModule().equals("oracle")) {
                    throw new LetException(Codez.ErrBadParameters, "aix 只支持 oracle 恢复");
                }

                final String clientID = body.getClientID();
                klog.info("recover task:{} start...", clientID);

                final OraclePodSpec oracle = pod.getSpec().getModule().getOracle();
                String target;
                String store;

                // 恢复开始前先挂载磁盘
                PodDiskStatus disk = pod.getStatus().getPodDisk();
                switch (disk.getKind()) {
                    case "AIXRaw":
                        String diskGroup = disk.getAixRaw().getGroup();
                        target = disk.getTarget();
                        store = disk.getStore();
                        SystemUtil.mkdir(store);

                        try {
                            new LBKit.diskgroup(letOptions.getScripts(), diskGroup, oracle.getAsm().getSid(), null)
                                    .User(oracle.getAsm().getUser())
                                    .Home(oracle.getAsm().getHome())
                                    .Listen(oracle.getAsm().getListen())
                                    .connect();
                        } catch (RuntimeException e) {
                            throw new LetException(Codez.ErrConnectDiskGroup, e.getMessage());
                        }

                        String pfile = Paths.get(store, body.getOracle().getOsid() + "__backup.ora").toString();
                        String controlfile = Paths.get(store, "controlfile.trace").toString();
                        if (!body.getOracle().hasFile()) {
                            throw new LetException(Codez.ErrLetNotExists, "missing oracle file");
                        }

                        try {
                            String pfileContent = SystemUtil.base64Decode(body.getOracle().getFile().getPfile());
                            String controlContent = SystemUtil.base64Decode(body.getOracle().getFile().getControl());
                            SystemUtil.writeAll(pfile, pfileContent);
                            SystemUtil.writeAll(controlfile, controlContent);
                        } catch (UnsupportedEncodingException e) {
                            throw new LetException(Codez.ErrLetNotExists, "文件格式错误: " + e.getMessage());
                        } catch (IOException e) {
                            throw new LetException(Codez.ErrLetNotExists, "创建文件错误: " + e.getMessage());
                        }

                        break;
                    case "AIXFS":
                        String dev = disk.getAixFS().getPv();
                        target = disk.getTarget();
                        store = target;
                        try {
                            LBKit.fsDisk fsDisk = new LBKit.fsDisk(letOptions.getScripts(), target, dev);
                            fsDisk.connect();
                        } catch (RuntimeException e) {
                            throw new LetException(Codez.ErrConnectFileDisk, e.getMessage());
                        }

                        if (!new File(Paths.get(target, "linkBackup").toString()).exists()) {
                            throw new LetException(Codez.ErrMountUnknown, "未发现标志文件");
                        }

                        break;
                    default:
                        target = disk.getTarget();
                        store = target;
                }

                String db = oracle.getInstance().getDb();
                String sid = oracle.getInstance().getSid();
                String user = oracle.getInstance().getUser();
                String home = oracle.getInstance().getHome();
                String listen = oracle.getInstance().getListen();
                final String rlog = Paths.get(letOptions.getPointDir(), String.format("%s-recover.log", sid)).toString();
                try {
                    SystemUtil.truncate(rlog);
                } catch (IOException e) {
                    klog.warn(e.getMessage());
                }
                final String params = Paths.get(letOptions.getPointDir(), String.format("%s-params", sid)).toString();
                StringBuffer sb = new StringBuffer();
                if (body.getOracle().getTotalMemory() > 0) {
                    sb.append(String.format("export totalmemory=%d\n", body.getOracle().getTotalMemory()));
                } else {
                    if (body.getOracle().getSga() > 0) {
                        sb.append(String.format("export sgc=%d\n", body.getOracle().getSga()));
                    }
                    if (body.getOracle().getPga() > 0) {
                        sb.append(String.format("export pgc=%d\n", body.getOracle().getPga()));
                    }
                }
                if (body.getOracle().getProcesses() > 0) {
                    sb.append(String.format("export processes=%d\n", body.getOracle().getProcesses()));
                }
                FileWriter fw = null;
                File fd = new File(params);
                try {
                    if (!fd.exists()) {
                        fd.createNewFile();
                    }
                    fw = new FileWriter(fd);
                    fw.write(sb.toString());
                    fw.close();
                } catch (IOException e) {
                    klog.error(e.getMessage());
                }

                final Progress progress = new Progress();

                final Channel<Message> channel = new Channel<>(1);
                // 另起一个线程，用来发送备份数据
                final Thread senderT = new Thread() {
                    @Override
                    public void run() {
                        CloneOut out = CloneOut.newBuilder().setClientID(clientID).build();
                        Message msg = null;
                        boolean over = false;
                        while (true) {
                            if (this.isInterrupted()) {
                                break;
                            }

                            if (over) {
                                break;
                            }

                            if (isCancel()) {
                                break;
                            }

                            msg = channel.take();
                            if (msg == null) {
                                continue;
                            }

                            switch (msg.getPhase()) {
                                case "+start":
                                case "+continue":
                                    progress.onNext();
                                    out = out.toBuilder().setProcess(progress.progress()).setPhase(msg.getPhase()).build();
                                    send(out);
                                    break;
                                case "+failed":
                                case "+completed":
                                    String rlogData = null;
                                    try {
                                        String readData = SystemUtil.readAll(rlog);
                                        rlogData = SystemUtil.base64Encode(readData);
                                    } catch (Exception e) {
                                        klog.error("read rlog failed: {}", e.getMessage());
                                    }
                                    out = out.toBuilder()
                                            .setProcess(100)
                                            .setPhase(msg.getPhase())
                                            .setLog(rlogData)
                                            .setErr(msg.getError())
                                            .build();
                                    send(out);
                                    over = true;
                                    break;
                            }
                        }
                    }

                    private void send(CloneOut out) {
                        RecoverResponse response = RecoverResponse.newBuilder()
                                .setOut(out)
                                .setStatus(LBStatus.OK())
                                .build();
                        if (!isCancel()) {
                            responseObserver.onNext(response);
                        }
                    }
                };

                // 另起一个线程，用来确定恢复情况
                final Thread healthT = new Thread() {
                    @Override
                    public void run() {
                        while (!isInterrupted() && !isCancel()) {
                            channel.put(new Message("+continue", "", 0));
                            try {
                                sleep(10000);
                            } catch (InterruptedException e) {
                            }
                        }
                    }
                };

                String format = "yyyy-MM-dd..HH:mm:ss";
                long rtime = body.getOracle().getRtime() * 1000;
                if (rtime == 0) {
                    rtime = new Date().getTime();
                }
                String recoverTime = new SimpleDateFormat(format).format(new Date(rtime));

                LBKit.Recover recover = LBKit.Recover.RecoverBuilder(letOptions.getScripts())
                        .Op("create")
                        .Osid(body.getOracle().getOsid())
                        .Sid(sid)
                        .Target(target)
                        .Store(store)
                        .Timestamp(recoverTime)
                        .Rlog(rlog)
                        .Params(params)
                        .User(user)
                        .Home(home)
                        .Listen(listen);
                if (oracle.hasAsm()) recover.EnableASM();
                senderT.start();
                healthT.start();
                switch (body.getOracle().getOp()) {
                    case "recover":
                        klog.info("Pod <{}:{}> recover", pod.getMetadata().getNamespace(), pod.getMetadata().getName());
                        recover.Execute();

                        String result = LBKit.Sqlplus.SqlplusBuilder(letOptions.getScripts())
                                .Op("getDBStatus")
                                .Sid(sid)
                                .User(user)
                                .Home(home)
                                .Listen(listen)
                                .Execute();
                        if (!result.toLowerCase().contains("open")) {
                            healthT.interrupt();
                            channel.put(new Message("+failed", "oracle 恢复错误", 0));
                            break;
                        }

                        channel.put(new Message("+completed", "", 0));
                        healthT.interrupt();
                        break;
                    case "mount":
                        klog.info("Pod <{}:{}> mount", pod.getMetadata().getNamespace(), pod.getMetadata().getName());
                        klog.info(recover.Commit());
                        healthT.interrupt();
                        channel.put(new Message("+completed", "", 0));
                        break;
                    default:
                        channel.put(new Message("+failed", "无效的恢复操作", 0));
                        healthT.interrupt();
                        break;
                }

                senderT.interrupt();
            }

            @Override
            public void onNext(RecoverRequest request) {
                try {
                    this.Do(request.getPod(), request.getBody(), responseObserver);
                } catch (LetException ex) {
                    Status status = new LBStatus(ex).getStatus();
                    RecoverResponse response = RecoverResponse.newBuilder()
                            .setStatus(status)
                            .build();
                    responseObserver.onNext(response);
                    this.onCompleted();
                }
            }

            @Override
            public void onError(Throwable throwable) {
                responseObserver.onError(throwable);
            }

            @Override
            public void onCompleted() {
                setCancel(true);
                responseObserver.onCompleted();
            }
        };
    }
}
